# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

import array
import dataclasses
import datetime
import enum
import functools
import logging
import pickle
import types
from typing import TypeVar, Union
from enum import Enum

from pyfory import ENABLE_FORY_CYTHON_SERIALIZATION
from pyfory import Language
from pyfory.error import TypeUnregisteredError

from pyfory.serializer import (
    Serializer,
    Numpy1DArraySerializer,
    NDArraySerializer,
    PyArraySerializer,
    DynamicPyArraySerializer,
    NoneSerializer,
    BooleanSerializer,
    ByteSerializer,
    Int16Serializer,
    Int32Serializer,
    Int64Serializer,
    Float32Serializer,
    Float64Serializer,
    StringSerializer,
    DateSerializer,
    TimestampSerializer,
    BytesSerializer,
    ListSerializer,
    TupleSerializer,
    MapSerializer,
    SetSerializer,
    EnumSerializer,
    SliceSerializer,
    DataClassSerializer,
    DataClassStubSerializer,
    StatefulSerializer,
    ReduceSerializer,
    FunctionSerializer,
    ObjectSerializer,
    TypeSerializer,
    ModuleSerializer,
    MappingProxySerializer,
    MethodSerializer,
    UnsupportedSerializer,
    NativeFuncMethodSerializer,
    PickleBufferSerializer,
)
from pyfory.meta.metastring import MetaStringEncoder, MetaStringDecoder
from pyfory.meta.meta_compressor import DeflaterMetaCompressor
from pyfory.type import (
    TypeId,
    int8,
    int16,
    int32,
    int64,
    float32,
    float64,
    load_class,
    is_struct_type,
    record_class_factory,
)
from pyfory._fory import (
    DYNAMIC_TYPE_ID,
    # preserve 0 as flag for type id not set in TypeInfo`
    NO_TYPE_ID,
)
from pyfory.meta.typedef import TypeDef
from pyfory.meta.typedef_decoder import decode_typedef, skip_typedef
from pyfory.meta.typedef_encoder import encode_typedef

try:
    import numpy as np
except ImportError:
    np = None

logger = logging.getLogger(__name__)
namespace_decoder = MetaStringDecoder(".", "_")
typename_decoder = MetaStringDecoder("$", "_")

if ENABLE_FORY_CYTHON_SERIALIZATION:
    from pyfory.serialization import TypeInfo
else:

    class TypeInfo:
        __slots__ = (
            "cls",
            "type_id",
            "serializer",
            "namespace_bytes",
            "typename_bytes",
            "dynamic_type",
            "type_def",
        )

        def __init__(
            self,
            cls: type = None,
            type_id: int = NO_TYPE_ID,
            serializer: Serializer = None,
            namespace_bytes=None,
            typename_bytes=None,
            dynamic_type: bool = False,
            type_def: TypeDef = None,
        ):
            self.cls = cls
            self.type_id = type_id
            self.serializer = serializer
            self.namespace_bytes = namespace_bytes
            self.typename_bytes = typename_bytes
            self.dynamic_type = dynamic_type
            self.type_def = type_def

        def __repr__(self):
            return f"TypeInfo(cls={self.cls}, type_id={self.type_id}, serializer={self.serializer})"

        def decode_namespace(self) -> str:
            if self.namespace_bytes is None:
                return ""
            return self.namespace_bytes.decode(namespace_decoder)

        def decode_typename(self) -> str:
            if self.typename_bytes is None:
                return ""
            return self.typename_bytes.decode(typename_decoder)


class TypeResolver:
    __slots__ = (
        "fory",
        "_metastr_to_str",
        "_type_id_counter",
        "_types_info",
        "_hash_to_metastring",
        "_metastr_to_type",
        "_hash_to_typeinfo",
        "_dynamic_id_to_typeinfo_list",
        "_dynamic_id_to_metastr_list",
        "_dynamic_write_string_id",
        "_dynamic_written_metastr",
        "_ns_type_to_typeinfo",
        "_named_type_to_typeinfo",
        "namespace_encoder",
        "namespace_decoder",
        "typename_encoder",
        "typename_decoder",
        "meta_compressor",
        "require_registration",
        "metastring_resolver",
        "language",
        "_type_id_to_typeinfo",
        "_meta_shared_typeinfo",
        "meta_share",
        "serialization_context",
        "_internal_py_serializer_map",
    )

    def __init__(self, fory, meta_share=False):
        self.fory = fory
        self.metastring_resolver = fory.metastring_resolver
        self.language = fory.language
        self.require_registration = fory.strict
        self._metastr_to_str = dict()
        self._metastr_to_type = dict()
        self._hash_to_metastring = dict()
        self._hash_to_typeinfo = dict()
        self._dynamic_written_metastr = []
        self._type_id_to_typeinfo = dict()
        self._type_id_counter = 64
        self._dynamic_write_string_id = 0
        # hold objects to avoid gc, since `flat_hash_map/vector` doesn't
        # hold python reference.
        self._types_info = dict()
        self._ns_type_to_typeinfo = dict()
        self._named_type_to_typeinfo = dict()
        self.namespace_encoder = MetaStringEncoder(".", "_")
        self.namespace_decoder = MetaStringDecoder(".", "_")
        # Cache for TypeDef and TypeInfo tuples (similar to Java's classIdToDef)
        self._meta_shared_typeinfo = {}
        self.typename_encoder = MetaStringEncoder("$", "_")
        self.typename_decoder = MetaStringDecoder("$", "_")
        self.meta_compressor = DeflaterMetaCompressor()
        self.meta_share = meta_share
        self._internal_py_serializer_map = {}

    def initialize(self):
        self._initialize_common()
        if self.fory.language == Language.PYTHON:
            self._initialize_py()
        else:
            self._initialize_xlang()
        self.serialization_context = self.fory.serialization_context

    def _initialize_py(self):
        register = functools.partial(self._register_type, internal=True)
        register(type(None), serializer=NoneSerializer)
        register(tuple, serializer=TupleSerializer)
        register(slice, serializer=SliceSerializer)
        register(np.ndarray, serializer=NDArraySerializer)
        register(array.array, serializer=DynamicPyArraySerializer)
        register(types.MappingProxyType, serializer=MappingProxySerializer)
        register(pickle.PickleBuffer, serializer=PickleBufferSerializer)
        if not self.require_registration:
            register(types.ModuleType, serializer=ModuleSerializer)
            self._internal_py_serializer_map = {
                ReduceSerializer: (self._stub_cls("__Reduce__"), self._next_type_id()),
                TypeSerializer: (self._stub_cls("__Type__"), self._next_type_id()),
                MethodSerializer: (self._stub_cls("__Method__"), self._next_type_id()),
                FunctionSerializer: (
                    self._stub_cls("__Function__"),
                    self._next_type_id(),
                ),
                NativeFuncMethodSerializer: (
                    self._stub_cls("__NativeFunction__"),
                    self._next_type_id(),
                ),
            }
            for serializer, (
                stub_cls,
                type_id,
            ) in self._internal_py_serializer_map.items():
                register(stub_cls, serializer=serializer, type_id=type_id)

    @staticmethod
    def _stub_cls(name: str):
        return record_class_factory(name, [])

    def _initialize_xlang(self):
        register = functools.partial(self._register_type, internal=True)
        register(array.array, type_id=DYNAMIC_TYPE_ID, serializer=DynamicPyArraySerializer)
        register(np.ndarray, type_id=DYNAMIC_TYPE_ID, serializer=NDArraySerializer)

    def _initialize_common(self):
        register = functools.partial(self._register_type, internal=True)
        register(None, type_id=TypeId.NA, serializer=NoneSerializer)
        register(bool, type_id=TypeId.BOOL, serializer=BooleanSerializer)
        register(int8, type_id=TypeId.INT8, serializer=ByteSerializer)
        register(int16, type_id=TypeId.INT16, serializer=Int16Serializer)
        register(int32, type_id=TypeId.INT32, serializer=Int32Serializer)
        register(int64, type_id=TypeId.INT64, serializer=Int64Serializer)
        register(int, type_id=TypeId.INT64, serializer=Int64Serializer)
        register(
            float32,
            type_id=TypeId.FLOAT32,
            serializer=Float32Serializer,
        )
        register(
            float64,
            type_id=TypeId.FLOAT64,
            serializer=Float64Serializer,
        )
        register(float, type_id=TypeId.FLOAT64, serializer=Float64Serializer)
        register(str, type_id=TypeId.STRING, serializer=StringSerializer)
        # TODO(chaokunyang) DURATION DECIMAL
        register(datetime.datetime, type_id=TypeId.TIMESTAMP, serializer=TimestampSerializer)
        register(datetime.date, type_id=TypeId.LOCAL_DATE, serializer=DateSerializer)
        register(bytes, type_id=TypeId.BINARY, serializer=BytesSerializer)
        for itemsize, ftype, typeid in PyArraySerializer.typecode_dict.values():
            register(
                ftype,
                type_id=typeid,
                serializer=PyArraySerializer(self.fory, ftype, typeid),
            )
        if np:
            # overwrite pyarray  with same type id.
            # if pyarray are needed, one must annotate that value with XXXArrayType
            # as a field of a struct.
            for dtype, (
                itemsize,
                format,
                ftype,
                typeid,
            ) in Numpy1DArraySerializer.dtypes_dict.items():
                register(
                    ftype,
                    type_id=typeid,
                    serializer=Numpy1DArraySerializer(self.fory, ftype, dtype),
                )
        register(list, type_id=TypeId.LIST, serializer=ListSerializer)
        register(set, type_id=TypeId.SET, serializer=SetSerializer)
        register(dict, type_id=TypeId.MAP, serializer=MapSerializer)
        try:
            import pyarrow as pa
            from pyfory.format.serializer import (
                ArrowRecordBatchSerializer,
                ArrowTableSerializer,
            )

            register(
                pa.RecordBatch,
                type_id=TypeId.ARROW_RECORD_BATCH,
                serializer=ArrowRecordBatchSerializer,
            )
            register(pa.Table, type_id=TypeId.ARROW_TABLE, serializer=ArrowTableSerializer)
        except Exception:
            pass

    def register_type(
        self,
        cls: Union[type, TypeVar],
        *,
        type_id: int = None,
        namespace: str = None,
        typename: str = None,
        serializer=None,
    ):
        return self._register_type(
            cls,
            type_id=type_id,
            namespace=namespace,
            typename=typename,
            serializer=serializer,
        )

    def _register_type(
        self,
        cls: Union[type, TypeVar],
        *,
        type_id: int = None,
        namespace: str = None,
        typename: str = None,
        serializer=None,
        internal=False,
    ):
        """Register type with given type id or typename. If typename is not None, it will be used for
        cross-language serialization."""
        if serializer is not None and not isinstance(serializer, Serializer):
            try:
                serializer = serializer(self.fory, cls)
            except BaseException:
                try:
                    serializer = serializer(self.fory)
                except BaseException:
                    serializer = serializer()
        n_params = len({typename, type_id, None}) - 1
        if n_params == 0 and typename is None:
            type_id = self._next_type_id()
        if n_params == 2:
            raise TypeError(f"type name {typename} and id {type_id} should not be set at the same time")
        if type_id not in {0, None}:
            # multiple type can have same tpe id
            if type_id in self._type_id_to_typeinfo and cls in self._types_info:
                raise TypeError(f"{cls} registered already")
        elif cls in self._types_info:
            raise TypeError(f"{cls} registered already")
        register_type = self._register_xtype if self.fory.language == Language.XLANG else self._register_pytype
        return register_type(
            cls,
            type_id=type_id,
            namespace=namespace,
            typename=typename,
            serializer=serializer,
            internal=internal,
        )

    def _register_xtype(
        self,
        cls: Union[type, TypeVar],
        *,
        type_id: int = None,
        namespace: str = None,
        typename: str = None,
        serializer=None,
        internal=False,
    ):
        if serializer is None:
            if issubclass(cls, enum.Enum):
                serializer = EnumSerializer(self.fory, cls)
                type_id = TypeId.NAMED_ENUM if type_id is None else ((type_id << 8) + TypeId.ENUM)
            else:
                serializer = None
                if self.meta_share:
                    type_id = TypeId.NAMED_COMPATIBLE_STRUCT if type_id is None else ((type_id << 8) + TypeId.COMPATIBLE_STRUCT)
                else:
                    type_id = TypeId.NAMED_STRUCT if type_id is None else ((type_id << 8) + TypeId.STRUCT)
        elif not internal:
            type_id = TypeId.NAMED_EXT if type_id is None else ((type_id << 8) + TypeId.EXT)

        return self.__register_type(
            cls,
            type_id=type_id,
            serializer=serializer,
            namespace=namespace,
            typename=typename,
            internal=internal,
        )

    def _register_pytype(
        self,
        cls: Union[type, TypeVar],
        *,
        type_id: int = None,
        namespace: str = None,
        typename: str = None,
        serializer: Serializer = None,
        internal: bool = False,
    ):
        # Set default type_id when None, similar to _register_xtype
        if type_id is None and typename is not None:
            type_id = self._next_type_id()
        return self.__register_type(
            cls,
            type_id=type_id,
            namespace=namespace,
            typename=typename,
            serializer=serializer,
            internal=internal,
        )

    def __register_type(
        self,
        cls: Union[type, TypeVar],
        *,
        type_id: int = None,
        namespace: str = None,
        typename: str = None,
        serializer: Serializer = None,
        internal: bool = False,
    ):
        dynamic_type = type_id is not None and type_id < 0
        # In metashare mode, for struct types, we want to keep serializer=None
        # so that _set_typeinfo will be called to create the TypeDef-based serializer
        should_create_serializer = (
            not internal
            and serializer is None
            and not (self.meta_share and typename is not None and type_id is not None and is_struct_type(type_id & 0xFF))
        )

        if should_create_serializer:
            serializer = self._create_serializer(cls)

        if typename is None:
            typeinfo = TypeInfo(cls, type_id, serializer, None, None, dynamic_type)
        else:
            if namespace is None:
                splits = typename.rsplit(".", 1)
                if len(splits) == 2:
                    namespace, typename = splits
            ns_metastr = self.namespace_encoder.encode(namespace or "")
            ns_meta_bytes = self.metastring_resolver.get_metastr_bytes(ns_metastr)
            type_metastr = self.typename_encoder.encode(typename)
            type_meta_bytes = self.metastring_resolver.get_metastr_bytes(type_metastr)
            typeinfo = TypeInfo(cls, type_id, serializer, ns_meta_bytes, type_meta_bytes, dynamic_type)
            self._named_type_to_typeinfo[(namespace, typename)] = typeinfo
            self._ns_type_to_typeinfo[(ns_meta_bytes, type_meta_bytes)] = typeinfo
        self._types_info[cls] = typeinfo
        if type_id is not None and type_id != 0 and (self.language == Language.PYTHON or not TypeId.is_namespaced_type(type_id)):
            if type_id not in self._type_id_to_typeinfo or not internal:
                self._type_id_to_typeinfo[type_id] = typeinfo
        self._types_info[cls] = typeinfo
        return typeinfo

    def _next_type_id(self):
        type_id = self._type_id_counter = self._type_id_counter + 1
        while type_id in self._type_id_to_typeinfo:
            type_id = self._type_id_counter = self._type_id_counter + 1
        return type_id

    def register_serializer(self, cls: Union[type, TypeVar], serializer):
        assert isinstance(cls, (type, TypeVar)), cls
        if cls not in self._types_info:
            raise TypeUnregisteredError(f"{cls} not registered")
        typeinfo = self._types_info[cls]
        if self.fory.language == Language.PYTHON:
            typeinfo.serializer = serializer
            return
        type_id = prev_type_id = typeinfo.type_id
        self._type_id_to_typeinfo.pop(prev_type_id)
        if typeinfo.serializer is not serializer:
            if typeinfo.typename_bytes is not None:
                type_id = typeinfo.type_id & 0xFFFFFF00 | TypeId.NAMED_EXT
            else:
                type_id = typeinfo.type_id & 0xFFFFFF00 | TypeId.EXT
        self._type_id_to_typeinfo[type_id] = typeinfo

    def get_serializer(self, cls: type):
        """
        Returns
        -------
            Returns or create serializer for the provided type
        """
        return self.get_typeinfo(cls).serializer

    def get_typeinfo(self, cls, create=True):
        type_info = self._types_info.get(cls)
        if type_info is not None:
            if type_info.serializer is None:
                self._set_typeinfo(type_info)
            return type_info
        elif not create:
            return None
        if self.require_registration and not issubclass(cls, Enum):
            raise TypeUnregisteredError(f"{cls} not registered")
        logger.info("Type %s not registered", cls)
        serializer = self._create_serializer(cls)
        type_id = None
        if self.language == Language.PYTHON:
            if isinstance(serializer, EnumSerializer):
                type_id = TypeId.NAMED_ENUM
            elif isinstance(serializer, (ObjectSerializer, StatefulSerializer)):
                type_id = TypeId.NAMED_EXT
            elif self._internal_py_serializer_map.get(type(serializer)) is not None:
                type_id = self._internal_py_serializer_map.get(type(serializer))[1]
            if not self.require_registration:
                if isinstance(serializer, DataClassSerializer):
                    type_id = TypeId.NAMED_STRUCT
        if type_id is None:
            raise TypeUnregisteredError(f"{cls} must be registered using `fory.register_type` API")
        return self.__register_type(
            cls,
            type_id=type_id,
            namespace=cls.__module__,
            typename=cls.__qualname__,
            serializer=serializer,
        )

    def _set_typeinfo(self, typeinfo):
        type_id = typeinfo.type_id & 0xFF
        if is_struct_type(type_id):
            if self.meta_share:
                type_def = encode_typedef(self, typeinfo.cls)
                if type_def is not None:
                    typeinfo.serializer = type_def.create_serializer(self)
                    typeinfo.type_def = type_def
                else:
                    # Fallback to regular serializer
                    typeinfo.serializer = DataClassSerializer(self.fory, typeinfo.cls, xlang=not self.fory.is_py)
            else:
                typeinfo.serializer = DataClassSerializer(self.fory, typeinfo.cls, xlang=not self.fory.is_py)
        else:
            typeinfo.serializer = self._create_serializer(typeinfo.cls)

        return typeinfo

    def _create_serializer(self, cls):
        for clz in cls.__mro__:
            type_info = self._types_info.get(clz)
            if type_info and type_info.serializer and type_info.serializer.support_subclass():
                serializer = type(type_info.serializer)(self.fory, cls)
                break
        else:
            if cls is types.FunctionType:
                # Use FunctionSerializer for function types (including lambdas)
                serializer = FunctionSerializer(self.fory, cls)
            elif dataclasses.is_dataclass(cls):
                # lazy create serializer to handle nested struct fields.
                serializer = DataClassStubSerializer(self.fory, cls, xlang=not self.fory.is_py)
            elif issubclass(cls, enum.Enum):
                serializer = EnumSerializer(self.fory, cls)
            elif ("builtin_function_or_method" in str(cls) or "cython_function_or_method" in str(cls)) and "<locals>" not in str(cls):
                serializer = NativeFuncMethodSerializer(self.fory, cls)
            elif cls is type(self.initialize):
                # Handle bound method objects
                serializer = MethodSerializer(self.fory, cls)
            elif issubclass(cls, type):
                # Handle Python type objects and metaclass such as numpy._DTypeMeta(i.e. np.dtype)
                serializer = TypeSerializer(self.fory, cls)
            elif cls is array.array:
                # Handle array.array objects with DynamicPyArraySerializer
                # Note: This will use DynamicPyArraySerializer for all array.array objects
                serializer = DynamicPyArraySerializer(self.fory, cls)
            elif (hasattr(cls, "__reduce__") and cls.__reduce__ is not object.__reduce__) or (
                hasattr(cls, "__reduce_ex__") and cls.__reduce_ex__ is not object.__reduce_ex__
            ):
                # Use ReduceSerializer for objects that have custom __reduce__ or __reduce_ex__ methods
                # This has higher precedence than StatefulSerializer and ObjectSerializer
                # Only use it for objects with custom reduce methods, not default ones from the object
                serializer = ReduceSerializer(self.fory, cls)
            elif hasattr(cls, "__getstate__") and hasattr(cls, "__setstate__"):
                # Use StatefulSerializer for objects that support __getstate__ and __setstate__
                serializer = StatefulSerializer(self.fory, cls)
            elif hasattr(cls, "__dict__") or hasattr(cls, "__slots__"):
                serializer = ObjectSerializer(self.fory, cls)
            else:
                # c-extension types will go to here
                serializer = UnsupportedSerializer(self.fory, cls)
        return serializer

    def is_registered_by_name(self, cls):
        typeinfo = self._types_info.get(cls)
        if typeinfo is None:
            return False
        return TypeId.is_namespaced_type(typeinfo.type_id & 0xFF)

    def is_registered_by_id(self, cls):
        typeinfo = self._types_info.get(cls)
        if typeinfo is None:
            return False
        return not TypeId.is_namespaced_type(typeinfo.type_id & 0xFF)

    def get_registered_name(self, cls):
        typeinfo = self._types_info.get(cls)
        assert typeinfo is not None, f"{cls} not registered"
        return typeinfo.decode_namespace(), typeinfo.decode_typename()

    def get_registered_id(self, cls):
        typeinfo = self._types_info.get(cls)
        assert typeinfo is not None, f"{cls} not registered"
        return typeinfo.type_id

    def _load_metabytes_to_typeinfo(self, ns_metabytes, type_metabytes):
        typeinfo = self._ns_type_to_typeinfo.get((ns_metabytes, type_metabytes))
        if typeinfo is not None:
            return typeinfo
        ns = ns_metabytes.decode(self.namespace_decoder)
        typename = type_metabytes.decode(self.typename_decoder)
        # the hash computed between languages may be different.
        typeinfo = self._named_type_to_typeinfo.get((ns, typename))
        if typeinfo is not None:
            self._ns_type_to_typeinfo[(ns_metabytes, type_metabytes)] = typeinfo
            return typeinfo
        cls = load_class(ns + "#" + typename)
        typeinfo = self.get_typeinfo(cls)
        self._ns_type_to_typeinfo[(ns_metabytes, type_metabytes)] = typeinfo
        return typeinfo

    def write_typeinfo(self, buffer, typeinfo):
        if typeinfo.dynamic_type:
            return
        # Check if meta share is enabled first
        if self.meta_share:
            self.write_shared_type_meta(buffer, typeinfo)
            return
        type_id = typeinfo.type_id
        internal_type_id = type_id & 0xFF
        buffer.write_varuint32(type_id)
        if TypeId.is_namespaced_type(internal_type_id):
            self.metastring_resolver.write_meta_string_bytes(buffer, typeinfo.namespace_bytes)
            self.metastring_resolver.write_meta_string_bytes(buffer, typeinfo.typename_bytes)

    def read_typeinfo(self, buffer):
        # Check if meta share is enabled first
        if self.meta_share:
            return self.read_shared_type_meta(buffer)

        type_id = buffer.read_varuint32()
        internal_type_id = type_id & 0xFF
        if TypeId.is_namespaced_type(internal_type_id):
            ns_metabytes = self.metastring_resolver.read_meta_string_bytes(buffer)
            type_metabytes = self.metastring_resolver.read_meta_string_bytes(buffer)
            typeinfo = self._ns_type_to_typeinfo.get((ns_metabytes, type_metabytes))
            if typeinfo is None:
                ns = ns_metabytes.decode(self.namespace_decoder)
                typename = type_metabytes.decode(self.typename_decoder)
                typeinfo = self._named_type_to_typeinfo.get((ns, typename))
                if typeinfo is not None:
                    self._ns_type_to_typeinfo[(ns_metabytes, type_metabytes)] = typeinfo
                    return typeinfo
                # TODO(chaokunyang) generate a dynamic type and serializer
                #  when meta share is enabled.
                name = ns + "." + typename if ns else typename
                raise TypeUnregisteredError(f"{name} not registered")
            return typeinfo
        else:
            return self._type_id_to_typeinfo[type_id]

    def get_typeinfo_by_id(self, type_id):
        """Get typeinfo by type_id."""
        return self._type_id_to_typeinfo[type_id]

    def get_typeinfo_by_name(self, namespace, typename):
        """Get typeinfo by namespace and typename."""
        return self._named_type_to_typeinfo.get((namespace, typename))

    def get_meta_compressor(self):
        return self.meta_compressor

    def write_shared_type_meta(self, buffer, typeinfo):
        """Write shared type meta information."""
        meta_context = self.fory.serialization_context.meta_context
        meta_context.write_shared_typeinfo(buffer, typeinfo)

    def read_shared_type_meta(self, buffer):
        """Read shared type meta information."""
        meta_context = self.serialization_context.meta_context
        assert meta_context is not None, "Meta context must be set when meta share is enabled"
        typeinfo = meta_context.read_shared_typeinfo(buffer)
        return typeinfo

    def write_type_defs(self, buffer):
        """Write all type definitions that need to be sent."""
        meta_context = self.fory.serialization_context.meta_context
        if meta_context is None:
            return
        writing_type_defs = meta_context.get_writing_type_defs()
        buffer.write_varuint32(len(writing_type_defs))
        for type_def in writing_type_defs:
            # Just copy the encoded bytes directly
            buffer.write_bytes(type_def.encoded)

    def read_type_defs(self, buffer):
        """Read all type definitions from the buffer."""
        meta_context = self.fory.serialization_context.meta_context
        if meta_context is None:
            return

        num_type_defs = buffer.read_varuint32()
        for i in range(num_type_defs):
            # Read the header (first 8 bytes) to get the type ID
            header = buffer.read_int64()
            # Check if we already have this TypeDef cached
            type_info = self._meta_shared_typeinfo.get(header)
            if type_info is not None:
                # Skip the rest of the TypeDef binary for faster performance
                skip_typedef(buffer, header)
            else:
                # Read the TypeDef and create TypeInfo
                type_def = decode_typedef(buffer, self, header=header)
                type_info = self._build_type_info_from_typedef(type_def)
                # Cache the tuple for future use
                self._meta_shared_typeinfo[header] = type_info
            meta_context.add_read_typeinfo(type_info)

    def _build_type_info_from_typedef(self, type_def):
        """Build TypeInfo from TypeDef using TypeDef's create_serializer method."""
        # Create serializer using TypeDef's create_serializer method
        serializer = type_def.create_serializer(self)
        ns_metastr = self.namespace_encoder.encode(type_def.namespace or "")
        ns_meta_bytes = self.metastring_resolver.get_metastr_bytes(ns_metastr)
        type_metastr = self.typename_encoder.encode(type_def.typename)
        type_meta_bytes = self.metastring_resolver.get_metastr_bytes(type_metastr)
        typeinfo = TypeInfo(
            type_def.cls,
            type_def.type_id,
            serializer,
            ns_meta_bytes,
            type_meta_bytes,
            False,
            type_def,
        )
        return typeinfo

    def reset(self):
        pass

    def reset_read(self):
        pass

    def reset_write(self):
        pass
